package com.heo.task;

import java.util.LinkedList;
import java.util.Queue;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import org.apache.log4j.Logger;
import com.heo.task.thread.CallBackInfo;
import com.heo.task.thread.CallBackThread;
import com.heo.util.Utils;

public class CallBackTaskQueue {

	private static Logger logger = Logger.getLogger(CallBackTaskQueue.class);

	// 线程池维护线程的最少数量
	private final static int CORE_POOL_SIZE = 10;

	// 线程池维护线程的最大数量
	private final static int MAX_POOL_SIZE = 300;

	// 线程池维护线程所允许的空闲时间
	private final static int KEEP_ALIVE_TIME = 0;

	// 线程池所使用的缓冲队列大小
	private final static int WORK_QUEUE_SIZE = 1000;
	
	//如果消息队列没有数据，就休眠，再次获取
	private final static int SLEEP_TIME = 100;
	
	//消息队列的最大长度，超过就丢弃
	private final static int MAX_QUEUE_SIZE = 10000;

	private static class SingletonHolder {
		private static final CallBackTaskQueue INSTANCE = new CallBackTaskQueue();
	}

	private CallBackTaskQueue() {
		new Thread(accessBufferThread).start();
	}

	public static final CallBackTaskQueue getInstance() {
		return SingletonHolder.INSTANCE;
	}
	
	public int getReceiveTaskSize() {
		return msgQueue.size();
	}

	// 消息缓冲队列
	Queue<CallBackInfo> msgQueue = new LinkedList<CallBackInfo>();

	final RejectedExecutionHandler handler = new RejectedExecutionHandler() {

		@Override
		public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
			CallBackInfo callBackInfo = ((CallBackThread) r).getCallBackInfo();
			logger.info(callBackInfo.toString() + "消息放入队列中重新等待执行");
			msgQueue.offer(callBackInfo);
		}
	};

	final ThreadPoolExecutor threadPool = new ThreadPoolExecutor(CORE_POOL_SIZE, MAX_POOL_SIZE, KEEP_ALIVE_TIME,
			TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(WORK_QUEUE_SIZE), this.handler);

	// 访问消息缓存的调度线程
	// 查看是否有待定请求，如果有，则创建一个新的CallBackThread，并添加到线程池中
	Runnable accessBufferThread = new Runnable() {
		@Override
		public void run() {
			try {
				logger.info("处理线程启动成功");
				while (true) {
					if (hasMoreAcquire()) {
						CallBackInfo callBackInfo = msgQueue.poll();
						logger.info("获取消息=" + callBackInfo.toString());
						if (Utils.checkNull(callBackInfo)) {
							Runnable task = new CallBackThread(callBackInfo);
							threadPool.execute(task);
						}else {
							logger.info("消息内容为空丢弃");
						}
					} else {
						Thread.sleep(SLEEP_TIME);
					}
				}
			} catch (InterruptedException e) {
				logger.error("处理线程异常" + e.getMessage());
			}
		}
	};

	private synchronized boolean hasMoreAcquire() {
		boolean bool = !msgQueue.isEmpty();
		if(bool) {
			CallBackInfo callBackInfo = msgQueue.peek();
			long sysTime = System.currentTimeMillis();
			if (Utils.checkNull(callBackInfo)) {
				if(callBackInfo.getReSendCount() <= callBackInfo.getMaxReSendCount()) {
					if(callBackInfo.getLastSendTime() == 0) {
						bool = true;
					}else if((callBackInfo.getLastSendTime() + callBackInfo.getExpireTime() > sysTime)) {
						if((callBackInfo.getLastSendTime() + callBackInfo.getSendIntervalTime()) <= sysTime) {
							bool = true;
						} else {
							bool = false;
						}
					}else {
						msgQueue.poll();
						logger.info("超过重发时间丢弃" + callBackInfo.toString());
						bool = false;
					}
				}else {
					msgQueue.poll();
					logger.info("超过重发次数丢弃" + callBackInfo.toString());
					bool = false;
				}
				
			}else {
				msgQueue.poll();
				logger.info("消息内容为空丢弃" + callBackInfo.toString());
				bool = false;
			}
		}
		return bool;
	}

	public boolean addCallBackMsg(CallBackInfo callBackInfo) {
		if(msgQueue.size() >= MAX_QUEUE_SIZE) {
			logger.info("添加回调失败，队列已满，请重试。");
			return false;
		}
		boolean bool = msgQueue.offer(callBackInfo);
		logger.info("添加消息=" + callBackInfo.toString() + ", 结果=" + bool);
		return bool;
	}
}
